package me.shiqiang.im.listeners;

import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.SyncMsg;
import me.shiqiang.im.msg.SyncReq;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cuishiqiang
 * @date 2018/1/12 下午11:08
 */
@Component("MsgSyncReqListener")
public class MsgSyncReqListener extends ConnectServerListener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public String eventName() {
        return MessageCmd.SYNC_MSG_REQ;
    }

    @Autowired
    StringRedisTemplate redisTemplate;

//    @Autowired
//    Socket socket;

//    AtomicInteger count = new AtomicInteger();
//
//    ConcurrentLinkedQueue<SyncReq> queue = new ConcurrentLinkedQueue<>();

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Override
    public void call(Object... objects) {
        JSONObject obj = (JSONObject) objects[0];
        long userId = obj.getLong("userId");
        long syncKey = obj.getLong("syncKey");
        SyncReq req = new SyncReq();
        req.setUserId(userId);
        req.setSyncKey(syncKey);
//        queue.add(req);
        queue.add(connection -> {
            Set<byte[]> bytes = connection
                    .zRangeByScore(String.valueOf(req.getUserId()).getBytes(),
                            req.getSyncKey() + 1, Long.MAX_VALUE, 0, 100);
            SyncMsg syncMsg = new SyncMsg(bytes.size());
            syncMsg.setUserId(req.getUserId());
            for (byte[] b : bytes) {
                try {
                    MsgBody body = JsonUtils.getInstance().fromJson(new String(b), MsgBody.class);
                    syncMsg.getMsgs().add(body);
                } catch (IOException e) {

                }
            }
            socket.emit(MessageCmd.SYNC_MSG_RES, JSONObject.wrap(syncMsg));
        });
//        int count = this.count.incrementAndGet();
//        if (count == 1000) {
//            this.count.getAndAdd(0 - count);
//            processReq(count);
//        }

    }

//    /**
//     * 每隔0.5秒，响应消息同步请求
//     */
//    @Scheduled(fixedRate = 500)
//    public void check() {
//        int count = this.count.get();
//        if (count > 0) {
//            this.count.getAndAdd(0 - count);
//            processReq(count);
//        }
//    }
//
//    @Async
//    public void processReq(int count) {
//        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
//            for (int i = 0; i < count; i++) {
//                SyncReq peek = queue.peek();
//                if (peek == null) {
//                    break;
//                }
//                Set<byte[]> bytes = connection
//                        .zRangeByScore(String.valueOf(peek.getUserId()).getBytes(),
//                                peek.getSyncKey() + 1, Long.MAX_VALUE, 0, 100);
//                SyncMsg syncMsg = new SyncMsg(bytes.size());
//                syncMsg.setUserId(peek.getUserId());
//                for (byte[] b : bytes) {
//                    try {
//                        MsgBody body = JsonUtils.getInstance().fromJson(new String(b), MsgBody.class);
//                        syncMsg.getMsgs().add(body);
//                    } catch (IOException e) {
//
//                    }
//                }
//                socket.emit(MessageCmd.SYNC_MSG_RES, syncMsg);
//            }
//            return null;
//        });
//    }
}
